package com.intct.func;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.intct.hbase.bean.DimConfigBean;
import com.intct.util.JDBCUtil;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @author gufg
 * @since 2025-08-20 09:53
 */
public class DimConnectFunc extends BroadcastProcessFunction<String, String, JSONObject> {

    private MapStateDescriptor mapStateDescriptor;

    //
    private Map<String, DimConfigBean> configMap = new HashMap<>();

    public DimConnectFunc(MapStateDescriptor mapStateDescriptor) {
        this.mapStateDescriptor = mapStateDescriptor;
    }

    // 生命Open，只在加载执行一次
    @Override
    public void open(Configuration parameters) throws Exception {
        Connection connection = JDBCUtil.getJDBCConnect();
        List<DimConfigBean> dimConfigBeanList =
                JDBCUtil.getQuery(connection, "select * from travel_config.dim_table_config", DimConfigBean.class);
        for (DimConfigBean dimConfigBean : dimConfigBeanList) {
            configMap.put(dimConfigBean.getSourceTable(), dimConfigBean);
        }
        if (connection != null) {
            connection.close();
        }
    }

    // 主流数据
    @Override
    public void processElement(String jsonStr,
                               BroadcastProcessFunction<String, String, JSONObject>.ReadOnlyContext ctx,
                               Collector<JSONObject> out) throws Exception {
        // 获取广播状态 --- 广播流数据
        ReadOnlyBroadcastState<String, DimConfigBean> broadcastState = ctx.getBroadcastState(mapStateDescriptor);

        // 从json string 转为json object --- 主流
        JSONObject jsonObj = JSON.parseObject(jsonStr);

        // 表名  -- 主流
        String table = jsonObj.getString("table");

        // 获取广播流中对应key值
        DimConfigBean dimConfigBean = broadcastState.get(table) != null ? broadcastState.get(table) : configMap.get(table);

        // 如果两流中匹配到，是维度表
        if (dimConfigBean != null) {

            // 列限定符 (列名)
            String sinkColumns = dimConfigBean != null ? dimConfigBean.getSinkColumns() : configMap.get(table).getSinkColumns();
            List<String> columnList = Arrays.asList(sinkColumns.split(","));

                /*
                    {"op":"r","after":
                        {"code":150125,
                        "name":"内蒙古自治区呼和浩特市武川县"
                        },
                     "table":"area_info"}
                 */

            // 处理主流与广播流数据，字段是否存
            JSONObject afterObj = jsonObj.getJSONObject("after");

            afterObj.entrySet().removeIf(field -> !columnList.contains(field.getKey()));

//            for (Map.Entry<String, Object> entry : afterObj.entrySet()) {
//                String key = entry.getKey();
//
//                if (!columnList.contains(key)) {
//                    afterObj.remove(key);
//                }
//            }

            // sink表名称
            String sinkTable =  dimConfigBean != null ? dimConfigBean.getSinkTable() : configMap.get(table).getSinkTable();

            // sink列族
            String sinkFamily = dimConfigBean != null ? dimConfigBean.getSinkFamily() : configMap.get(table).getSinkFamily();

            // rowkey  == 主键字段
            String sinkRowKey = dimConfigBean != null ? dimConfigBean.getSinkRowKey() : configMap.get(table).getSinkRowKey();

            // 操作类型
            String op = jsonObj.getString("op");


                    /*
                    {
                        "sinkTable":"table_name",
                        "sinkFamily": "Family",
                        "sinkRowkey": "Rowkey",
                        "op": "op",
                        "after": {
                            "code":"code",
                            "name":"name"
                        }
                    }
                     */
            JSONObject tmpObj = new JSONObject();
            tmpObj.put("sinkTable", sinkTable);
            tmpObj.put("sinkFamily", sinkFamily);
            tmpObj.put("sinkRowKey", sinkRowKey);
            tmpObj.put("op", op);
            tmpObj.put("after", afterObj);

            // 输出
            out.collect(tmpObj);
        }
    }

    // 广播流数据
    @Override
    public void processBroadcastElement(String jsonStr,
                                        BroadcastProcessFunction<String, String, JSONObject>.Context ctx,
                                        Collector<JSONObject> out) throws Exception {
        // {"op":"r","after":{
        // "sink_row_key":"code",
        // "sink_family":"info",
        // "sink_table":"dim_area_info",
        // "source_table":"area_info",
        // "sink_columns":"code,name"}
        // }
        // 获取广播状态
        BroadcastState broadcastState = ctx.getBroadcastState(mapStateDescriptor);

        // 从json string 转为json object
        JSONObject jsonObj = JSON.parseObject(jsonStr);

        // 操作类型
        String op = jsonObj.getString("op");

        JSONObject afterObj = jsonObj.getJSONObject("after");

        // sink table做为map state的主键
        String key = afterObj.getString("source_table");

        // 构造配置JavaBean
        DimConfigBean dimConfigBean = DimConfigBean.builder()
                .sourceTable(afterObj.getString("source_table"))
                .sinkTable(afterObj.getString("sink_table"))
                .sinkFamily(afterObj.getString("sink_family"))
                .sinkRowKey(afterObj.getString("sink_row_key"))
                .sinkColumns(afterObj.getString("sink_columns"))
                .build();

        // 如果是删除配置表数据
        if ("d".equals(op)) {
            // 删除
            broadcastState.remove(key);
            configMap.remove(key);
        } else {
            // 增加、修改、查询
            broadcastState.put(key, dimConfigBean);
            configMap.put(key, dimConfigBean);
        }
    }
}
